package com.song.storm_demo.wordcount_01;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class WordCouter extends BaseBasicBolt {
	
	private Map<String,Integer> couters = new HashMap<String, Integer>();
	private volatile boolean edit = false;
	
	
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		// TODO Auto-generated method stub
		super.prepare(stormConf, context);
		new Thread(new Runnable() {
			
			public void run() {
				while(true)
				{
					if(edit)
					{
						for(Entry<String, Integer> entry :couters.entrySet())
						{
							System.out.println(entry.getKey() +"---------"+entry.getValue());
						}
						System.out.println("wordcounter --------------------------------");
						edit = false;
					}
					
					try {
						Thread.sleep(2* 1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				
			}
		}).start();
	}

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		 String str = tuple.getString(0);
		 if(!couters.containsKey(str))
			 couters.put(str, 1);
		 else{
			 Integer c = couters.get(str)+1;
			 couters.put(str, c);
		 }
		 edit = true;
		 System.out.println("wordcounter +++++++++++++++++++++++++++++++");
	}

	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub

	}

}
